package com.atguigu.flink.sql.query;

import com.atguigu.flink.pojo.WaterSensor;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Schema;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * Created by 黄凯 on 2023/6/26 0026 20:32
 *
 * @author 黄凯
 * 永远相信美好的事情总会发生.
 *
 * Flink SQL的窗口:
 *  *    1. 分组窗口
 *  *         滚动窗口( 时间 、计数)
 *  *         滑动窗口( 时间 、计数)
 *  *         会话窗口( 时间)
 *  *
 *  *         TableAPI定义窗口语法:
 *  *            .window(Tumble.over(lit(10).minutes()).on($("rowtime")).as("w"));
 *  *         TableAPI使用窗口语法:
 *  *            Table table = input
 *  *              .window([GroupWindow w].as("w"))  // 定义窗口并指定别名为 w
 *  *              .groupBy($("w"))  // 以窗口 w 对表进行分组
 *  *              .select($("b").sum());  // 聚合
 *  *        SQL中使用语法:
 *  *            SELECT
 *  *                user,
 *  *                TUMBLE_START(order_time, INTERVAL '1' DAY) AS wStart,
 *  *                SUM(amount) FROM Orders
 *  *           GROUP BY
 *  *              TUMBLE(order_time, INTERVAL '1' DAY),
 *  *              user
 *  *
 *  *
 *  *    2. Window TVF(窗口表值函数)
 *  *         滚动窗口
 *  *         滑动窗口
 *  *         累积窗口
 *  *         会话窗口(暂不支持)
 *  *
 *  *         SQL中使用语法:
 *  *           SELECT
 *  *              id,
 *  *              vc,
 *  *              ts,
 *  *              window_start,
 *  *              window_end
 *  *           FROM TABLE(
 *  *              TUMBLE(TABLE Bid, DESCRIPTOR(bidtime), INTERVAL '10' MINUTES)
 *  *           )
 *  *           group by window_start ,window_end ;
 *  *
 *  *    3. Over(开窗)
 */
public class Flink04_WindowTVFSQl {

    public static void main(String[] args) {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        SingleOutputStreamOperator<WaterSensor> ds = env.socketTextStream("127.0.0.1", 8888)
                .map(
                        line -> {
                            String[] fields = line.split(",");
                            return new WaterSensor(fields[0].trim(), Long.valueOf(fields[1].trim()), Integer.valueOf(fields[2].trim()));
                        }

                );
        //创建表环境
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
        //流转表
        Schema schema = Schema.newBuilder()
                .column("id" , "STRING")
                .column("vc" , "INT")
                .column("ts" ,"BIGINT")
                .columnByExpression("pt" , "PROCTIME()")
                .columnByExpression("et" , "TO_TIMESTAMP_LTZ(ts,3)")
                .watermark("et" , "et - INTERVAL '1' SECOND")
                .build();
        Table table = tableEnv.fromDataStream(ds, schema);
        table.printSchema();
        tableEnv.createTemporaryView("t1" , table);

        //分组窗口
        //滚动
        //处理时间滚动窗口
        String sql1 =
                " select " +
                        "   id , " +
                        "   sum(vc) ," +
                        "   window_start ," +
                        "   window_end " +
                        " from TABLE( " +
                        "   TUMBLE( TABLE t1 , DESCRIPTOR(pt) , INTERVAL '5' SECOND) " +
                        " ) " +
                        " group by window_start ,window_end , id " ;

        //事件时间滚动窗口

        String sql2 =
                " select " +
                        "   id , " +
                        "   sum(vc) ," +
                        "   window_start ," +
                        "   window_end " +
                        " from TABLE( " +
                        "   TUMBLE( TABLE t1 , DESCRIPTOR(et) , INTERVAL '5' SECOND) " +
                        " ) " +
                        " group by window_start ,window_end , id " ;

        //滑动

        //处理时间滑动窗口
        String sql3 =
                " select " +
                        "   id , " +
                        "   sum(vc) ," +
                        "   window_start ," +
                        "   window_end " +
                        " from TABLE( " +
                        "   HOP( TABLE t1 , DESCRIPTOR(pt) , INTERVAL '5' SECOND , INTERVAL '10' SECOND ) " +
                        " ) " +
                        " group by window_start ,window_end , id " ;

        //事件时间滑动窗口
        String sql4 =
                " select " +
                        "   id , " +
                        "   sum(vc) ," +
                        "   window_start ," +
                        "   window_end " +
                        " from TABLE( " +
                        "   HOP( TABLE t1 , DESCRIPTOR(et) , INTERVAL '5' SECOND , INTERVAL '10' SECOND ) " +
                        " ) " +
                        " group by window_start ,window_end , id " ;

        //累积
        // 例如: 统计一天的日活， 每小时输出一次结果
        //  第一次统计: [0点 - 1点]
        //  第二次统计: [0点 - 2点]
        //  第三次统计: [0点 - 3点]
        // ........
        //  第24次统计:[0点  - 24点]
        // 当完成统计的范围，重置。

        //处理时间累积窗口
        String sql5 =
                " select " +
                        "   id , " +
                        "   sum(vc) ," +
                        "   window_start ," +
                        "   window_end " +
                        " from TABLE( " +
                        "   CUMULATE( TABLE t1 , DESCRIPTOR(pt) , INTERVAL '2' SECOND , INTERVAL '10' SECOND ) " +
                        " ) " +
                        " group by window_start ,window_end , id " ;

        //事件时间累积窗口
        String sql6 =
                " select " +
                        "   id , " +
                        "   sum(vc) ," +
                        "   window_start ," +
                        "   window_end " +
                        " from TABLE( " +
                        "   CUMULATE( TABLE t1 , DESCRIPTOR(et) , INTERVAL '2' SECOND , INTERVAL '10' SECOND ) " +
                        " ) " +
                        " group by window_start ,window_end , id " ;

        //使用窗口
        tableEnv.sqlQuery(sql6).execute().print();

        try {
            env.execute();
        } catch (Exception e) {
            throw new RuntimeException(e);
        }

    }

}
